package disruptor;

import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.Executors;

/**
 * @author: 李德才
 * @description:
 * @create: 2020-12-20 21:09
 **/
public class Disruptor_MultiConsumer {


    /**
     * 创建 Disruptor
     */
    public static Disruptor<DisruptorEvent> createDisruptor() {
        return new Disruptor<>(new DisruptorEventFactory(), 1024, Executors.defaultThreadFactory(),
                ProducerType.MULTI, new SleepingWaitStrategy());
    }


    /**
     * handleEventsWith 和 handleEventsWithWorkerPool
     * 对于某一条消息m，handleEventsWith方法返回的EventHandlerGroup，Group中的每个消费者都会对m进行消费，各个消费者之间不存在竞争。
     * handleEventsWithWorkerPool方法返回的EventHandlerGroup，Group的消费者对于同一条消息m不重复消费；也就是，如果c0消费了消息m，则c1不再消费消息m
     */

    /**
     * handleEventsWith 可重复消费
     */
    public static void handleEventsWith() {
        Disruptor<DisruptorEvent> disruptor = createDisruptor();
        //        5 组消费者
        for (int i = 0; i < 5; i++) {
            EventHandlerGroup<DisruptorEvent> handlerGroup = disruptor.handleEventsWith(new DisruptorEventHandler(1));
            /**
             * 支持多个参数
             */
            handlerGroup.handleEventsWith(new DisruptorEventHandler(2), new DisruptorEventHandler(3));
            /**
             * 支持后续处理，这里执行清理程序
             * 如果不清理，Disruptor 将空间使用完毕后将会从头开始覆盖
             */
            handlerGroup.then(new DisruptorEvent());
        }
        publishEvent(disruptor);
    }

    /**
     * handleEventsWithWorkerPool 不可重复消费
     */
    public static void handleEventsWithWorkerPool() {
        Disruptor<DisruptorEvent> disruptor = createDisruptor();
        //        5 组消费者
        for (int i = 0; i < 5; i++) {
            EventHandlerGroup<DisruptorEvent> handlerGroup = disruptor.handleEventsWithWorkerPool(new DisruptorWorkHandler(1));
            handlerGroup.handleEventsWithWorkerPool(new DisruptorWorkHandler(2));
            handlerGroup.then(new DisruptorEvent());
        }
        publishEvent(disruptor);
    }

    /**
     * 多个消费者顺序消费，消费顺序为  1,2,3
     */
    public static void handleEventsWithList() {
        Disruptor<DisruptorEvent> disruptor = createDisruptor();
        DisruptorEventHandler disruptorEventHandler1 = new DisruptorEventHandler(1);
        DisruptorEventHandler disruptorEventHandler2 = new DisruptorEventHandler(2);
        DisruptorEventHandler disruptorEventHandler3 = new DisruptorEventHandler(3);
        EventHandlerGroup<DisruptorEvent> handlerGroup = disruptor.handleEventsWith(disruptorEventHandler1);
        disruptor.after(disruptorEventHandler1).then(disruptorEventHandler2);
        disruptor.after(disruptorEventHandler2).then(disruptorEventHandler3);
        /**
         * 支持后续处理，这里执行清理程序
         * 如果不清理，Disruptor 将空间使用完毕后将会从头开始覆盖
         */
        handlerGroup.then(new DisruptorEvent());
        publishEvent(disruptor);
    }

    /**
     * 发布事件
     */
    public static void publishEvent(Disruptor<DisruptorEvent> disruptor) {
        disruptor.start();
        disruptor.publishEvent(((disruptorEvent, Long) -> disruptorEvent.set(1L)));
        disruptor.shutdown();
    }

    public static void main(String[] args) {
//        多消费者重复消费
//        handleEventsWith();
//        多消费者不重复消费
//        handleEventsWithWorkerPool();
//        多消费者顺序消费
        handleEventsWithList();
    }
}
